package com.zxjbyte.yiyi.framework.cdc.mysql;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MYSQL binlog连接器生成的MYSQL binlog事件的事件监听器包装器。
 *
 * @Author zhangxingjia
 * @Date 2025/4/11 15:13
 * @Version: 1.0
 */
@Slf4j
public class BinlogEventListener implements BinaryLogClient.EventListener {

    private final AtomicBoolean stopNow = new AtomicBoolean(false);
    private static final int QUEUE_OFFER_TIMEOUT_MSEC = 100;

    private final BlockingQueue<RawBinlogEvent> queue;
    private final BinaryLogClient client;

    public BinlogEventListener(BinaryLogClient client, BlockingQueue<RawBinlogEvent> q) {
        this.client = client;
        this.queue = q;
    }

    /**
     * 创建并返回与指定binlog客户端和事件队列关联的BinlogEventListener实例。
     * @param client
     * @param q 传递事件的队列
     * @return
     */
    public static BinlogEventListener createBinlogEventListener(BinaryLogClient client, BlockingQueue<RawBinlogEvent> q) {
        return new BinlogEventListener(client, q);
    }

    public void start() {
        stopNow.set(false);
    }

    public void stop() {
        stopNow.set(true);
    }


    @Override
    public void onEvent(Event event) {
        RawBinlogEvent ep = new RawBinlogEvent(event, client.getBinlogFilename());
        try {
            while (!stopNow.get()) {
                if (queue.offer(ep, QUEUE_OFFER_TIMEOUT_MSEC, TimeUnit.MILLISECONDS)) {
                    return;
                }
            }

            log.info(">>>> [cdc-mysql] Stopped while waiting to enqueue event");
        } catch (InterruptedException e) {
            log.warn(">>>> [cdc-mysql] Interrupted while adding event to the queue", e);
        }
    }
}
